package com.shujia.flink.sql

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row

object Demo11FLinOnHbase {
  def main(args: Array[String]): Unit = {

    val configuration = new Configuration

    //网络链接超时时间
    configuration.setString("akka.ask.timeout", "100000")

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.createLocalEnvironment(1, configuration)


    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val bsEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)



    bsEnv.executeSql(
      """
        |
        |CREATE TABLE hTable (
        | id INT,
        | info ROW<name string,age int ,gender string,clazz string>,
        | PRIMARY KEY (id) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'student',
        | 'zookeeper.quorum' = 'master:2181'
        |)
        |
      """.stripMargin)


    bsEnv.executeSql(
      """
        |
        |CREATE TABLE clazz_num (
        | clazz String,
        | info ROW<num bigint>,
        | PRIMARY KEY (clazz) NOT ENFORCED
        |) WITH (
        | 'connector' = 'hbase-1.4',
        | 'table-name' = 'clazz_num',
        | 'zookeeper.quorum' = 'master:2181'
        |)
        |
      """.stripMargin)




    bsEnv.executeSql(
      """
        |insert into clazz_num
        |
        |select
        | clazz,ROW(num) as info
        | from
        |(
        |select info.clazz as clazz,count(1) as num from hTable group by info.clazz
        |) as a
        |
      """.stripMargin)




  }

}
